// Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: dserver support type discovery
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: support config log level by cmd tool
// feature: add statictic lookup
// ------------------------------------------------------------------

#include <edds/dds/builtin/lookup/LookupReplyListener.hpp>

#include <history/ParticipantListener.h>
#include <edds/rtps/builtin/BuiltinProtocols.h>

#include <ertps/types/TypeObjectFactory.h>
#include <ertps/types/TypeObject.h>
#include <ertps/rtps/reader/RTPSReader.h>
#include <history/WriterHistory.h>
#include <ertps/rtps/history/ReaderHistory.h>
#include <elog/Log.hpp>
#include <rtps/participant/RTPSParticipantImpl.h>

#include <edds/dds/builtin/lookup/LookupManager.hpp>

using evbs::ertps::rtps::RTPSReader;
using evbs::ertps::rtps::RTPSWriter;
using vbs::common::CacheChange_t;
using vbsutil::elog::Log;

using evbs::ertps::rtps::c_EntityId_Lookup_reply_writer;

using namespace evbs::ertps::types;

namespace evbs {
namespace edds {
namespace dds {
namespace builtin {

LookupReplyListener::LookupReplyListener(LookupManager* manager)
    : tlm_(manager), factory_(TypeObjectFactory::get_instance()) {}

LookupReplyListener::~LookupReplyListener() {}

void LookupReplyListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                const CacheChange_t* const changeIN) {
    logDebug(LOOKUP_SERVICE, "Received LookupReplyListener.");
    CacheChange_t* change = const_cast<CacheChange_t*>(changeIN);
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    if (change->writerGUID.entityId != c_EntityId_Lookup_reply_writer) {
        elogWarning(LOOKUP_SERVICE, "Received data from a bad endpoint.");
        (void)reader->getHistory()->remove_change(change);
        return;
    }

    // 1.4.0版本及后续版本和1.3.0版本的类型发现不再兼容
    const auto& version = tlm_->participant_->get_version_by_guid(change->writerGUID.guidPrefix);
    if (version[0] == 1 && version[1] == 3 && version[2] == 0) {
        elogWarning(LOOKUP_SERVICE,
                    "It is no longer compatible with version " << version[0] << "." << version[1] << "." << version[2]);
        (void)reader->getHistory()->remove_change(change);
        return;
    }

    Lookup_Reply reply;
    uint16_t continue_recvpkt_flag = 0;
    if (tlm_->recv_reply(*change, reply, continue_recvpkt_flag)) {
        if (reply.header.requestId.writer_guid() != tlm_->get_builtin_request_writer_guid()) {
            // This message isn't for us.
            (void)reader->getHistory()->remove_change(change);
            return;
        }

        switch (reply.data._d()) {
            case Statistic_getWRInfo_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->wr_infos_.find(sampe_id) != tlm_->wr_infos_.end()) {
                    if (continue_recvpkt_flag == 1U) {
                        tlm_->wr_infos_[sampe_id].infos.push_back(reply.data.getWriterReaderInfo().infos_);
                        break;
                    }
                    tlm_->wr_infos_[sampe_id].infos.push_back(reply.data.getWriterReaderInfo().infos_);
                    tlm_->wr_infos_[sampe_id].wr_promisers_.set_value(tlm_->wr_infos_[sampe_id].infos);
                    tlm_->wr_infos_.erase(sampe_id);
                }
            } break;
            case Statistic_getMatchGuids_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->guid_promises_.find(sampe_id) != tlm_->guid_promises_.end()) {
                    tlm_->guid_promises_[sampe_id].set_value(reply.data.getMatchGuids().guids_);
                    tlm_->guid_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_getSendInfo_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->send_promises_.find(sampe_id) != tlm_->send_promises_.end()) {
                    tlm_->send_promises_[sampe_id].set_value(reply.data.getSendInfo().info_);
                    tlm_->send_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_getRecvInfo_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->recv_promises_.find(sampe_id) != tlm_->recv_promises_.end()) {
                    tlm_->recv_promises_[sampe_id].set_value(reply.data.getRecvInfo().info_);
                    tlm_->recv_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_getWriterReaderQos_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->writer_qos_promises_.find(sampe_id) != tlm_->writer_qos_promises_.end()) {
                    tlm_->writer_qos_promises_[sampe_id].set_value(reply.data.getWriterQos().qos_);
                    tlm_->writer_qos_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_getMsgBrief_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->msg_brief_promises_.find(sampe_id) != tlm_->msg_brief_promises_.end()) {
                    tlm_->msg_brief_promises_[sampe_id].set_value(reply.data.getMsgBreifInfo().guid_msg_briefs_);
                    tlm_->msg_brief_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_getProxyInfo_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->proxy_info_promises_.find(sampe_id) != tlm_->proxy_info_promises_.end()) {
                    tlm_->proxy_info_promises_[sampe_id].set_value(reply.data.getProxyInfos().proxy_infos_);
                    tlm_->proxy_info_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_remoteConfig_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->remote_config_promises_.find(sampe_id) != tlm_->remote_config_promises_.end()) {
                    tlm_->remote_config_promises_[sampe_id].set_value(reply.data.remoteConfig().result_);
                    tlm_->remote_config_promises_.erase(sampe_id);
                }
            } break;
            case Statistic_getDServerInfo_Hash: {
                const evbs::ertps::rtps::SampleIdentity& sampe_id = reply.header.requestId;
                std::lock_guard<std::mutex> lock(tlm_->promise_lock_);
                if (tlm_->dserver_info_promises_.find(sampe_id) != tlm_->dserver_info_promises_.end()) {
                    tlm_->dserver_info_promises_[sampe_id].set_value(reply.data.dserverInfo().infos_);
                    tlm_->dserver_info_promises_.erase(sampe_id);
                }
            } break;
            case Lookup_getTypeInfo_Hash: {
                const Lookup_getTypeInfo_Result typeInfo = reply.data.getTypeInfo();
                {
                    std::lock_guard<std::mutex> types_guard(tlm_->types_lock_);
                    auto it =
                        std::find(tlm_->types_to_register_.begin(), tlm_->types_to_register_.end(), typeInfo.type_name);
                    if (it == tlm_->types_to_register_.end()) {
                        (void)reader->getHistory()->remove_change(change);
                        return;
                    }
                    tlm_->types_to_register_.erase(it);
                }
                tlm_->participant_->getListener()->on_type_information_received(change->writerGUID, typeInfo.type_name,
                                                                                typeInfo.information);
                logDebug(LOOKUP_SERVICE, tlm_->get_builtin_request_writer_guid()
                                             << " Received type info reply. " << typeInfo.type_name);
                break;
            }
            case Lookup_getTypes_Hash: {
                const Lookup_getTypes_Result types = reply.data.getType();
                // If build_dynamic_type failed, just sent the nullptr already contained on it.
                tlm_->participant_->getListener()->on_type_discovery(tlm_->participant_->getUserRTPSParticipant(),
                                                                     reply.header.requestId, types.types);
                logDebug(LOOKUP_SERVICE, tlm_->get_builtin_request_writer_guid() << " Received types reply.");
                break;
            }
            case Lookup_getDependencies_Hash: {
                tlm_->participant_->getListener()->on_type_dependencies_reply(
                    change->writerGUID, tlm_->participant_->getUserRTPSParticipant(), reply.header.requestId,
                    reply.data.getTypeDependencies().dependent_typeids);
                logDebug(LOOKUP_SERVICE, tlm_->get_builtin_request_writer_guid()
                                             << " Received type dependencies reply.");
                break;
            }
            default:
                break;
        }
    }
    (void)reader->getHistory()->remove_change(change);
}

void LookupReplyListener::onWriterChangeReceivedByAll(vbs::dispatcher::SubstrateWriter* const /*writer*/,
                                                      CacheChange_t* change) {
    tlm_->builtin_request_writer_history_->remove_change(change);
}

}  // namespace builtin
}  // namespace dds
}  // namespace edds
}  // namespace evbs
